package output;

import java.io.*;
import java.util.*;

public class SerialObjectStream extends Thread implements iObjectWriter<Serializable> {

	private int MAX_BUFFER = 100000; // 100,000 Magic Number
	
	private boolean keepGoing;
	private String filename;
	private ObjectOutputStream out;
	private Queue<Serializable> q;
	private int flushCount = 0;
	private char currentVolume;
	
	public boolean readyToWrite() {
		return keepGoing;
	}
	
	public String getFileName() {
		return filename;
	}

	public SerialObjectStream(String filename) {
		super("writer_" + filename);
		if (filename == null)
			throw new IllegalArgumentException();
		try {
			currentVolume = 'a';
			out = new ObjectOutputStream(new FileOutputStream(filename.replaceAll("\\.", currentVolume + "\\.")));
			this.filename = filename;
			q = new LinkedList<Serializable>();
		} catch (IOException ex) {
			throw new IllegalArgumentException();
		}
	}
	
	public boolean add(Serializable object) {
		boolean res = false;
		synchronized (q) {
			if (q.size() < MAX_BUFFER) { 
				q.offer(object);
				q.notifyAll();
				res = true;
			}
		}
		if (!res)
			System.err.println("\tRejected...");
		return res;
	}
	
	@Override
	public void run() {
		if (keepGoing)
			return;
		keepGoing = true;
		boolean notEmpty = true;
		Serializable temp = null;
		while(keepGoing || notEmpty) {
			synchronized (q) {
				while (!q.isEmpty())
					writeT(q.poll());
				notEmpty = !q.isEmpty();
			}
		}
		System.out.println("Closing stream: " + filename);
		if (out != null) {
			try {
				out.close();
			} catch (IOException e) {
			}
		}
	}

	public void kill() {
		keepGoing = false;
	}

	private void writeT(Serializable object) {
		if (object == null || out == null)
			return;
		try {
			out.writeObject(object);
			if (flushCount++ > 500000) {
				System.out.println("Switching stream: " + filename);
				currentVolume++;
				try {
					out.close();
					out = new ObjectOutputStream(new FileOutputStream(filename.replaceAll("\\.", currentVolume + "\\.")));
				} catch (IOException e) {
				}
				flushCount = 0;
			}
		} catch (IOException ex) {
		}
	}

	@Override
	public boolean write(Serializable obj) {
		return add(obj);
	}

	@Override
	public boolean unique(Serializable obj) {
		throw new UnsupportedOperationException();
	}
}
